package org.luxor.common.netty;

import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.core.thread.ThreadUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.luxor.common.netty.config.ClientPipelineConfig;
import org.luxor.common.netty.config.properties.NettyClientProperties;
import org.luxor.common.netty.protocol.handler.WatchdogClientHandler;
import org.luxor.commons.core.utils.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty客户端
 *
 * @author Mr.yan @date 2023/8/29
 */
public class TcpNioClient extends Thread {
    private final Logger log = LoggerFactory.getLogger(this.getClass());
    private boolean isInActive = false;

    /**
     * 当前网络通道
     */
    private ChannelFuture channelFuture;

    /**
     * 工作线程组
     */
    private EventLoopGroup workerGroup;

    /**
     * 客户端配置
     */
    protected NettyClientProperties properties;

    /**
     * 处理器链配置
     */
    private final ChannelInitializer<Channel> channelInitializer;

    /**
     * 重连处理器
     */
    private final WatchdogClientHandler watchdogHandler;

    /**
     * 统计连接次数
     */
    private final AtomicInteger connectCount;

    public TcpNioClient(ClientPipelineConfig config) {
        super("netty-client-worker-1");
        this.properties = config.clientProperties();
        this.channelInitializer = config;
        this.connectCount = new AtomicInteger(0);
        this.watchdogHandler = new WatchdogClientHandler(this);
    }

    public NettyClientProperties getClientOption() {
        return properties;
    }

    @Override
    public void run() {
        while (!isInActive && connectCount.get() < properties.getReconnectCountMax()) {
            isInActive = channelFuture != null && channelFuture.channel().isActive();
            // 等待连接间隔时间，越来越长
            int timeout = 2 << connectCount.get();
            ThreadUtils.sleepQuietly(timeout * 1000L);
            // 统计连接次数
            connectCount.incrementAndGet();
            try {
                log.info("[{}] The {}th try connection {}:{} Begin...", properties.getId(), connectCount, properties.getHost(), properties.getPort());
                doConnect();
            } catch (Exception e) {
                log.error("[{}] The {}th try connection {}:{} Failed.", properties.getId(), connectCount, properties.getHost(), properties.getPort(), e);
            }
        }
    }

    /**
     * 建立连接
     *
     * @return io.netty.channel.ChannelFuture
     */
    private void doConnect() throws InterruptedException {
        //关闭之前的资源
        close();

        AtomicBoolean isComplete = new AtomicBoolean(false);
        try {
            // 工作线程组
            workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, new NamedThreadFactory("netty-client-worker-", true));
            // 配置客户端启动参数
            Bootstrap bootstrap = new Bootstrap().group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(32 * 1024, 64 * 1024))
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, properties.getConnectTimeoutSeconds() * 1000)
                    .handler(this.watchdogHandler)
                    .handler(this.channelInitializer);
            // 发起异步连接操作
            channelFuture = bootstrap.connect(properties.getHost(), properties.getPort()).sync();
            // 监听结果
            channelFuture.addListener(cf -> {
                boolean succeed = cf.isSuccess();
                if (succeed) {
                    log.info("[{}] The {}th try connection {}:{} Succeed.", properties.getId(), connectCount.get(), properties.getHost(), properties.getPort());
                    // 重置连接次数
                    this.connectCount.set(0);
                } else {
                    log.error("[{}] The {}th try connection {}:{} Failed.", properties.getId(), connectCount.get(), properties.getHost(), properties.getPort());
                    channelFuture.channel().pipeline().fireChannelInactive();
                }
            });
            while (!isComplete.get()) {
                ThreadUtil.sleep(500, TimeUnit.NANOSECONDS);
            }
            isComplete.set(true);

            // 同步阻塞，等待客户端链路关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * 关闭连接
     */
    public void close() {
        try {
            if (channelFuture != null) {
                channelFuture.channel().close();
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
        } catch (Throwable e) {
            log.error("[{}] Client connector close error.", properties.getId(), e);
        }
    }
}
